package com.example.consumer.scheduled;

import com.example.consumer.config.RedisStreamConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Range;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.List;

/**
 * @Author: AdobePeng
 * @Date: 2021/3/11 14:59
 */
@Component
public class ScheduleJob {
    static final Logger LOGGER = LoggerFactory.getLogger(ScheduleJob.class);
    @Autowired
    StringRedisTemplate stringRedisTemplate;
    @Autowired
    RedisStreamConfig redisStreamConfig;

    /**
     * 每隔5秒钟，扫描一下有没有等待自己消费的
     * 主要消费那些转组过来的消息，如果转组次数大于1，则进行尝试消费
     */
    @Scheduled(cron="0/5 * * * * ?")
    public void reportCurrentTime() {
        StreamOperations<String, String, String> streamOperations = this.stringRedisTemplate.opsForStream();
        /*从消费者的pending队列中读取消息，能够进到这里面的，一定是非业务异常，例如接口超时、服务器宕机等。
        对于业务异常，例如字段解析失败等，丢进异常表或者redis*/
        PendingMessages pendingMessages = streamOperations.pending(redisStreamConfig.getStream(), Consumer.from(redisStreamConfig.getGroup(), redisStreamConfig.getConsumer()));
        if(pendingMessages.size() > 0){
            pendingMessages.forEach( pendingMessage -> {
                // 最后一次消费到现在的间隔
                Duration elapsedTimeSinceLastDelivery = pendingMessage.getElapsedTimeSinceLastDelivery();
                // 转组次数
                long totalDeliveryCount = pendingMessage.getTotalDeliveryCount();
                if(totalDeliveryCount > 1){
                    try{
                        RecordId id = pendingMessage.getId();
                        List<MapRecord<String, String, String>> result = streamOperations.range(redisStreamConfig.getStream(), Range.rightOpen(id.toString(),id.toString()));
                        MapRecord<String, String, String> entries = result.get(0);
                        // 消费消息
                        LOGGER.info("消费了消息id={}, 消息value={}", entries.getId(), entries.getValue());
                        // 手动ack消息
                        streamOperations.acknowledge(redisStreamConfig.getGroup(), entries);
                    }catch (Exception e){
                        // 异常处理
                        e.printStackTrace();
                    }
                }
            });
        }
    }
}
